package org.yonggan.shop.sql.report

import org.apache.spark.sql.DataFrame
import org.yonggan.shop.constant.ConfigurationManager
import org.yonggan.shop.utils.{SQLUtils, SparkUtils}

/**
  * 用户的复购率
  * 需求：数据什么样的，最终的结果什么样的？  时间范围 6月份  25%  白金会员
  * 步骤：
  *1.数据采集、预处理
  *2.购买两次以上的用户数量 / 购买商品的用户数量
  *3.数据存储到MySQL或其他存储系统中（设计表时要根据查询维度设计）
  */
object RptRePurchaseRate {

  def main(args: Array[String]): Unit = {

    val spark = SparkUtils.getSparkSession("用户的复购率")
    val sc = spark.sparkContext
    sc.setLogLevel("WARN")

    // 文件输出
    val jsonDF: DataFrame = spark.read.json(ConfigurationManager.TASK_INJSON)
    //    jsonDF.createTempView("v_data")

    val month = "2018-05"

    // 过滤出结算和支付成功的记录
    val tempDF = jsonDF.filter(
      s"""
         |event_type = 4
         |and openid is not null
         |and time is not null
      """.stripMargin).cache()

    tempDF.createTempView("v_order")

    //    tempDF.show()

    // 总的订单数据
    val total = spark.sql(
      """
        |select ctn from ( select count(distinct(openid)) as ctn from v_order)
      """.stripMargin)

    //    val sum = total.collect()(0).getLong(0)

    //购买用户总量
    //    val totalBst: Broadcast[Long] = sc.broadcast[Long](sum)

    //复购的次数
    //    val minPt = 2
    //    val maxPt = 3

    // 复购用户
    //    val sql =
    //      s"""
    //         |SELECT current_date() , count(1)/${totalBst.value.toDouble} as rpr FROM
    //         |(SELECT openid , COUNT(1) as ctn  FROM v_order WHERE pay_status = 1
    //         |GROUP BY openid HAVING ctn <= ${maxPt}  and ctn >= ${minPt})
    //      """.stripMargin
    //
    //    val resDf = SQLUtils.executor(sql, spark)
    //    resDf.show()

    /*-------------- 统计用户等级 ------------*/
    val sql =
      """
        |SELECT openid , COUNT(1) as ctn  FROM v_order WHERE pay_status = 1
        |GROUP BY openid HAVING ctn >= 2
      """.stripMargin

    SQLUtils.executor(sql, spark).createTempView("v_buy_lev")

    val levRDD =
      """
        |SELECT COUNT(1) counts,
        |CASE WHEN ctn <= 3 THEN 'lv1'
        |     WHEN ctn <= 5 and  ctn >=4 THEN 'lv2'
        |     ELSE  'lv3' END  level
        |  FROM v_buy_lev GROUP BY  level
      """.stripMargin

    val reslutLevRDD = SQLUtils.executor(levRDD, spark)
//    +------+-----+
//    |counts|level|
//    +------+-----+
//    |     4|  lv1|
//    |     2|  lv3|
//    |     1|  lv2|
//    +------+-----+

    reslutLevRDD.createTempView("v_lev_res")
    //行转列

    // 第一种方式 行转列
    val res =
      """
        |SELECT current_date(),
        |SUM(CASE WHEN level = 'lv1' THEN counts  ELSE 0 END ) LV1,
        |SUM(CASE WHEN level = 'lv2' THEN counts  ELSE 0 END ) LV2,
        |SUM(CASE WHEN level = 'lv3' THEN counts  ELSE 0 END ) LV3
        |FROM v_lev_res
      """.stripMargin

    SQLUtils.executor(res,spark).show()
//    +--------------+---+---+---+
//    |current_date()|LV1|LV2|LV3|
//    +--------------+---+---+---+
//    |    2018-06-11|  4|  1|  2|
//    +--------------+---+---+---+

    spark.stop
  }

}
